package com.hexiang.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.utils.MessageUtil;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

@Slf4j
@Component
public class RpcConsumer {


    private static final String CONSUMER_GROUP = "rpc-hello-topic-consumer-group";
    private static final String PRODUCER_GROUP = "rpc-hello-topic-producer-group";
    private static final String NAME_SRV_ADDR = "127.0.0.1:9876";
    private static final String TOPIC_NAME = "RPC_HELLO_TOPIC";

    private DefaultMQPushConsumer consumer;

    @PostConstruct
    public void init() throws MQClientException {
        // create a producer to send reply message
        DefaultMQProducer replyProducer = new DefaultMQProducer(PRODUCER_GROUP);
        replyProducer.setNamesrvAddr(NAME_SRV_ADDR);
        replyProducer.start();

        // create consumer
        consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
        consumer.setNamesrvAddr(NAME_SRV_ADDR);
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        // recommend client configs
        consumer.setPullTimeDelayMillsWhenException(0L);

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                try {
                    log.info("handle message: {}", new String(msg.getBody()));
                    String replyTo = MessageUtil.getReplyToClient(msg);
                    byte[] replyContent = "reply message contents.".getBytes();
                    // create reply message with given util, do not create reply message by yourself
                    Message replyMessage = MessageUtil.createReplyMessage(msg, replyContent);

                    // send reply message with producer
                    SendResult replyResult = replyProducer.send(replyMessage, 3000);
                    log.info("reply to {} , {}", replyTo, replyResult.toString());
                } catch (MQClientException | RemotingException | MQBrokerException e) {
                    log.error("consumeMessageErr", e);
                } catch (InterruptedException e) {
                    log.error("consumeMessageErr", e);
                    Thread.currentThread().interrupt();
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.subscribe(TOPIC_NAME, "*");
        consumer.start();
    }

    @PreDestroy
    public void shutdown() {
        consumer.shutdown();
    }
}
